/* 05-prod-cons.c

   Producer-consumer co-routining.  There are two producers and two consumers.
   A successor produces puts consecutive numbers into its stream.  A times
   consumer multiplies all consumed numbers by 5 and puts the results into
   its stream.  A consumer prints tokens from the times stream.  This 
   illustrates that producer-consumer relationships can be formed into 
   complex networks.
*/
#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>
#include <stdbool.h>

int idcnt = 1;

/* One of these per Stream
   Holds:  the mutex lock and notifier condition variables
           a 'value' taken from a producer
           a variable 'available' indicating a token can be consumed
           a structure with information regarding the processing of tokens 
           an identity
*/
typedef struct stream_struct {
   struct stream_struct *next;
   pthread_mutex_t lock;
   pthread_cond_t notifier;
   void *value;
   bool available;
   void *args;             /* arguments for the producing stream */
   int id;
} Stream;

/* prod: a linked list of streams that a producer consumes tokens from
   self: the producer's output stream  */
typedef struct {
   Stream *prod, *self;
} Args;

/* return 'value' which should have a value from a call to put */
void *get(void *stream) {
   void *ret;

   bool available = ((Stream*)stream)->available;
   pthread_mutex_t *lock = &((Stream*)stream)->lock;
   pthread_cond_t *notifier = &((Stream*)stream)->notifier;
   
   pthread_mutex_lock(lock);    /* lock other threads out of this section    */
   if (!available)              /* if nothing in the buffer, wait and open   */
      pthread_cond_wait(notifier, lock);     /* the section to other threads */
   ((Stream*)stream)->available = false;     /* the token has been consumed  */
   ret = ((Stream*)stream)->value;           /* the token to be consumed     */
   pthread_cond_signal(notifier);  /* wake up sleeping producer, if any      */
   pthread_mutex_unlock(lock);  /* unlock the section */

   return ret;
}

/* 'value' is the value to move to the consumer */
void put(void *stream, void *value) {
   bool available = ((Stream*)stream)->available;
   pthread_mutex_t *lock = &((Stream*)stream)->lock;
   pthread_cond_t *notifier = &((Stream*)stream)->notifier;
   
   pthread_mutex_lock(lock);       /* lock the section */
   if (available)                  /* if a token is waiting for consumption  */
      pthread_cond_wait(notifier, lock);   /* put this thread to sleep       */
   ((Stream*)stream)->available = true;  /* a token is ready for consumption */
   ((Stream*)stream)->value = value;  /* this is the token to be consumed    */
   pthread_cond_signal(notifier);     /* wake up a sleeping consumer, if any */
   pthread_mutex_unlock(lock);     /* unlock the section */

   return;
}

/* Put 1,2,3,4,5... into the self stream */
void *successor (void *streams) {
   Stream *self = ((Args*)streams)->self;
   int i;
   
   for (i=1 ; ; i++) {
      printf("Successor: sending %d\n",i);
      put(self, (void*)i);
      printf("Successor: sent %d\n",i);
   }
   pthread_exit(NULL);
}

/* multiply all tokens from the self stream by (int)self->args and insert
   the resulting tokens into the self stream */
void *times (void *streams) {
   Stream *self = ((Args*)streams)->self;
   Stream *prod = ((Args*)streams)->prod;
   int i, value, in;
   
   printf("Times(%d) connected to Successor (%d)\n", self->id, prod->id);
   while (true) {
      printf("\tTimes (%d): request next token from successor %d\n",
	     self->id, prod->id);
      int in = (int)get(prod);
      value = in * (int)(self->args);
      printf("\tTimes (%d): got %d from successor %d, sending %d\n", 
	     self->id, in, prod->id, value);
      put(self, (void*)value);
      printf("\tTimes: sent %d\n", value);
   }
   pthread_exit(NULL);
}

/* Final consumer in the network */
void *consumer (void *streams) {
   Stream *prod = ((Args*)streams)->prod;
   int i, value;
   
   for (i=0 ; i < 10 ; i++) {
      printf("\t\t\t\t\t\tConsumer: requests next token\n");
      printf("\t\t\t\t\t\tConsumer: got %d\n", (int)get(prod));
   }
   
   pthread_exit(NULL);
}

/* initialize streams */
void init_stream (Args *args, Stream *self, void *data) {
   if (self != NULL) {
      self->available = false;
      self->value = NULL;
      self->next = NULL;
      self->args = data;
      self->id = idcnt++;
      pthread_mutex_init(&self->lock, NULL);
      pthread_cond_init(&self->notifier, NULL);
   }
   args->self = self;
   args->prod = NULL;
}

/* puts an initialized stream object onto the end of a stream's input list */
void connect (Args *arg, Stream *s) {  
   s->next = arg->prod;
   arg->prod = s;
}

int main () {
   pthread_t suc, tms, con;
   Stream succ_stream, times_stream;
   Args succ_args, times_args, cons_args;
   pthread_attr_t attr;

   /* initialize successor stream */
   init_stream(&succ_args, &succ_stream, NULL); 
   /* initialize times stream */
   init_stream(&times_args, &times_stream, (void*)5);
   /* connect times stream to successor stream */
   connect(&times_args, &succ_stream);
   /* initialize consumer stream */
   init_stream(&cons_args, NULL, NULL);
   /* connect consumer stream to times stream */
   connect(&cons_args, &times_stream);

   pthread_attr_init(&attr);
   pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
   pthread_create(&suc, &attr, successor, (void*)&succ_args);
   pthread_create(&tms, &attr, times, (void*)&times_args);
   pthread_create(&con, &attr, consumer, (void*)&cons_args);

   pthread_join(con, NULL);    /* cancel all running threads when the */
                               /* is finished consuming */
   pthread_cancel(tms);
   pthread_cancel(suc);

   pthread_exit(NULL);
}
   
